package com.ms.service.impl;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.ms.context.BaseContext;
import com.ms.domain.entity.Message;
import com.ms.mapper.MessageMapper;
import com.ms.service.MessageService;
import com.ms.websocket.WebSocketServer;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;


@Service
@RequiredArgsConstructor
@Slf4j
public class MessageServiceImpl implements MessageService {
    private final RedisTemplate<String, Object> redisTemplate;
    private final ObjectMapper objectMapper;
    private final MessageMapper messageMapper;

    // 生成Redis全局唯一ID
    private Long nextId() {
        return redisTemplate
                .opsForValue()
                .increment("global:msg_id");
    }

    /**
     * 发送没有关联资源的消息
     *
     * @param userId  用户ID
     * @param title   消息标题
     * @param type    消息类型
     * @param content 消息内容
     */
    @Override
    public void sendMessage(Integer userId, String title, String type, String content) {
        sendMessage(userId, title, type, content, null);
    }

    /**
     * 发送完整的消息
     *
     * @param userId    用户ID
     * @param title     消息标题
     * @param content   消息内容
     * @param type      消息类型
     * @param relatedId 关联ID
     */
    @Override
    public void sendMessage(Integer userId, String title, String type, String content, Integer relatedId) {
        Message message = Message.builder()
                .id(nextId())
                .userId(userId)
                .title(title)
                .content(content)
                .type(type)
                .relatedId(relatedId)
                .isRead(false)
                .createTime(System.currentTimeMillis())
                .build();

        // 添加到消息列表（Hash）
        String messageKey = "message:" + message.getId();
        // 处理relatedId为null的情况
        String relatedIdStr = message.getRelatedId() != null ? String.valueOf(message.getRelatedId()) : "";
        redisTemplate.opsForHash().putAll(messageKey, Map.of(
                "id", String.valueOf(message.getId()),
                "userId", message.getUserId(),
                "title", message.getTitle(),
                "content", message.getContent(),
                "type", message.getType(),
                "relatedId", relatedIdStr,
                "isRead", message.getIsRead() ? "1" : "0",
                "createTime", String.valueOf(message.getCreateTime())
        ));

        // 添加到用户消息列表（Sorted Set）
        String userMessageKey = "user:" + userId + ":messages";
        redisTemplate.opsForZSet().add(
                userMessageKey,
                message.getId(),
                message.getCreateTime()
        );

        // 更新未读数
        redisTemplate.opsForValue().increment("unread_message_count:" + userId);

        // websocket推送
        Map<String, Object> pushData = Map.of(
                "event", "NEW_MESSAGE",
                "data", Map.of(
                        "id", message.getId(),
                        "title", title,
                        "content", content,
                        "type", type,
                        "timestamp", message.getCreateTime(),
                        "unread", true
                )
        );

        // 使用JSON库（如Jackson/Gson）
        try {
            String jsonMessage = objectMapper.writeValueAsString(pushData);
            WebSocketServer.sendToUser(userId, jsonMessage);
        } catch (JsonProcessingException e) {
            throw new RuntimeException(e);
        }

        log.info("发送消息给用户 {}: {} - {}", userId, title, content);
    }

    /**
     * 分页获取当前用户的消息列表
     *
     * @param userId 用户ID
     * @return 未读消息数量
     */
    @Override
    public List<Message> getUserMessages(Integer userId, int pageNum, int pageSize) {
        String userMessageKey = "user:" + userId + ":messages";
        int start = (pageNum - 1) * pageSize;
        int end   = start + pageSize - 1;

        // 倒序取消息ID
        Set<Object> messageIds = redisTemplate.opsForZSet()
                .reverseRange(userMessageKey, start, end);
        if (messageIds == null || messageIds.isEmpty()) {
            return List.of();
        }

        // 批量获取消息内容（保持顺序）
        List<Message> messages = new ArrayList<>(messageIds.size());
        redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
            for (Object id : messageIds) {
                connection.hashCommands().hGetAll(("message:" + id).getBytes(StandardCharsets.UTF_8));
            }
            return null;
        }).forEach(obj -> {
            Map<Object, Object> hash = (Map<Object, Object>) obj;
            Message msg = Message.builder()
                    .id(parseLong(String.valueOf(hash.getOrDefault("id", 0))))
                    .userId(parseInt(String.valueOf(hash.getOrDefault("userId", 0))))
                    .title(String.valueOf(hash.getOrDefault("title", "")))
                    .content(String.valueOf(hash.getOrDefault("content", "")))
                    .type(String.valueOf(hash.getOrDefault("type", "")))
                    .isRead("1".equals(String.valueOf(hash.getOrDefault("isRead", "0"))))
                    .relatedId(parseInt(String.valueOf(hash.getOrDefault("relatedId", 0))))
                    .createTime(parseLong(String.valueOf(hash.getOrDefault("createTime", 0))))
                    .build();
            messages.add(msg);
        });

        return messages;

    }


    /**
     * 获取当前用户的未读消息数量
     *
     * @param userId 用户ID
     * @return 未读消息数量
     */
    @Override
    public Integer getUnreadMessageCount(Integer userId) {
        String key = "unread_message_count:" + userId;
        Object count = redisTemplate.opsForValue().get(key);
        if (count == null) {
            // 从数据库统计未读消息数量
            return messageMapper.countUnreadMessages(userId);
        }
        try {
            return Integer.parseInt(count.toString());
        } catch (NumberFormatException e) {
            log.error("解析未读消息数量失败", e);
            redisTemplate.delete(key); // 删除异常数据
            return messageMapper.countUnreadMessages(userId);
        }
    }

    @Override
    public void markAsRead(Integer messageId) {
        String messageKey = "message:" + messageId;
        Map<Object, Object> messages = redisTemplate.opsForHash().entries(messageKey);

        if (messages.get("isRead").equals("0")) {
            redisTemplate.opsForHash().put(messageKey, "isRead", "1");
            // 未读消息数减一
            int userId = Integer.parseInt(messages.get("userId").toString());
            redisTemplate.opsForValue().decrement("unread_message_count:" + userId);
        }
    }

    /**
     * 批量标记所有消息为已读
     *
     * @param userId 用户ID
     */
    @Override
    public void markAllAsRead(Integer userId) {
        String userMessageKey = "user:" + userId + ":messages";
        Set<Object> messageIds = redisTemplate.opsForZSet().range(userMessageKey, 0, -1);
        if (messageIds != null) {
            for (Object messageId : messageIds) {
                markAsRead(Integer.parseInt(messageId.toString()));
            }
        }
    }

    /**
     * 删除消息
     *
     * @param id
     */
    @Override
    public void deleteMessage(Integer id) {
        Integer currentId = BaseContext.getCurrentId();
        String userMessageKey = "user:" + currentId + ":messages";
        String messageKey = "message:" + id;
        log.info("用户 {}删除消息 {}", currentId, id);
        redisTemplate.delete(messageKey);
        redisTemplate.opsForZSet().remove(userMessageKey, id);
    }

    /**
     * 批量删除所有消息
     */
    @Override
    public void deleteAllMessages() {
        Integer currentId = BaseContext.getCurrentId();
        String userMessageKey = "user:" + currentId + ":messages";
        Set<Object> messageIds = redisTemplate.opsForZSet().range(userMessageKey, 0, -1);
        log.info("用户 {} 删除所有消息", currentId);
        if (messageIds != null) {
            for (Object messageId : messageIds) {
                String messageKey = "message:" + messageId;
                redisTemplate.delete(messageKey);
            }
        }
        // 清理用户消息列表
        redisTemplate.delete(userMessageKey);
        // 清理未读消息计数
        redisTemplate.delete("unread_message_count:" + currentId);
    }

    /**
     * 安全解析Long
     *
     * @param str 字符串
     * @return 解析后的Long值
     */
    private long parseLong(String str) {
        if (str == null || str.trim().isEmpty()) return 0L;
        try {
            return Long.parseLong(str.trim());
        } catch (NumberFormatException e) {
            return 0L;
        }
    }

    /**
     * 安全解析Integer
     *
     * @param str 字符串
     * @return 解析后的Integer值
     */
    private int parseInt(String str) {
        if (str == null || str.trim().isEmpty()) return 0;
        try {
            return Integer.parseInt(str.trim());
        } catch (NumberFormatException e) {
            return 0;
        }
    }

    /**
     * 安全解析String
     *
     * @param bytes 字节数组
     * @return 解析后的String值
     */
    private String parseStr(byte[] bytes) {
        return bytes == null ? "" : new String(bytes, StandardCharsets.UTF_8);
    }
}
